package com.shujia.spark.sql

import org.apache.spark.sql.{Column, DataFrame, SparkSession}

object Demo6Score {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
      .appName("dsl")
      .master("local")
      .config("spark.sql.shuffle.partitions", 1)
      .getOrCreate()
    import spark.implicits._
    import org.apache.spark.sql.functions._

    val scoresDF: DataFrame = spark
      .read
      .format("csv")
      .schema("`姓名` STRING,`语文` DOUBLE,`数学` DOUBLE,`英语` DOUBLE")
      .option("sep", ",")
      .load("data/scores.txt")

    val score_map: Column = map(
      expr("'语文'"), $"语文",
      expr("'数学'"), $"数学",
      expr("'英语'"), $"英语")

    val table1: DataFrame = scoresDF
      .select($"姓名", explode(score_map) as Array("cname", "score"))

    table1.show()

    table1
      .groupBy($"姓名")
      .agg(
        sum(when($"cname" === "语文", $"score").otherwise(0)) as "语文",
        sum(when($"cname" === "数学", $"score").otherwise(0)) as "数学",
        sum(when($"cname" === "英语", $"score").otherwise(0)) as "英语",
      ).show()

  }
}
